作 者： jus tcodeit | 来源： 数据社
目录
处理器主频和散热遇到瓶颈，多核处理器成为主流，并行化计算应用不断增加。 互联网技术的发展让大多数企业能够积累大量的数据，而企业需要灵活快速地从这些数据中提取出有价值的信息来服务用户或帮助企业自身决策。然而处理器的主频和散热遇到了瓶颈，CPU难以通过纵向优化来提升性能，所以多核这种横向扩展成为了主流。也因此，开发者需要利用多核甚至分布式架构技术来提高企业的大数据处理能力。这些技术随着开源软件的成功而在业界得到广泛应用。
下面我稍微介绍一些大数据应用中通常出现的一些原理或者说是特征。 分布式：将数据分布到不同的节点（机器），从而存储大量的数据。而分布式同时为并行读写和计算提供了基础，这样就能提高数据的处理能力。 为什么不直接使用分布式的关系型数据库，比如主从模式的mysql？这主要是效率的问题。分布式关系型数据库为了实现分布式事务、线性一致性、维护自身索引结构等功能会对性能造成影响。而正如刚刚背景所提到，大数据需求重点是快速处理大量数据，帮助用户和企业的决策。这个决策就包括推荐、监控、数据分析等。这些场景并不一定需要数据库这种严格的约束，是OLAP而非OLTP。所以大数据技术会通过解除这些限制而提升性能。 批量处理：单位是上百MB的数据块而非一条条数据，这样在数据读写时能够整体操作，减少IO寻址的时间消耗。 最终线性一致性：大数据技术很多都放弃线性一致性，这主要是跨行/文档（关系型模型叫行，文档型模型叫文档）时非原子操作，在一行/一个文档内还是会做到原子的。为了读写性能而允许跨行/文档出现读写延迟。 增加数据冗余：规范化的数据能够减少数据量，但在使用时需要关联才能获得完整数据，而在大数据下进行多次关联的操作是十分耗时的。为此，一些大数据应用通过合并宽表减少关联来提高性能。 列式存储：读取数据时只读取业务所关心的列而不需要把整行数据都取出再做进行截取，而且列式的压缩率更高，因为一列里一般都是同类的数据。 副本：大数据存储通常都会有副本设置，这样即便其中一份出现丢失，数据也能从副本找到。 高可用：大数据应用通常都会考虑高可用，即某个节点挂了，会有其他的节点来继续它的工作。 由于这个分享会的标题起得有点大，包括存储、搜索、计算三大块，而且篇幅有限，所以我就只根据这三块中我了解且比较流行的开源组件来分享，而且只讲解大概的原理。毕竟下面的每个组件的原理和实战都可以各自出一本五六百页的书了，而且还没涉及源码细节的。下面首先来介绍分布式文件系统，就是把我们单台计算机的文件系统扩展到多台。 HDFS（Hadoop Distributed File System） 架构原理
图中有8台机器或者容器，两个client、5个DataNode、1个NameNode。一个分布式文本系统，组成：NameNode、DataNode和secondary namenode NameNode：作为master，管理元数据，包括文件名、副本数、数据块存储的位置，响应client的请求，接收workers的heartbeating和blockreport。 DataNode：管理数据（data block，存储在磁盘，包括数据本身和元数据）和处理master、client端的请求。定期向namenode发送它们所拥有的块的列表。 secondary namenode：备用master Block：默认128MB，但小于一个block的文件只会占用相应大小的磁盘空间。设置100+MB是为了尽量减少寻址时间占整个数据读取时间的比例，但如果block过大，又不适合数据的分散存储或计算。将数据抽象成block，还有其他好处，比如分离元数据和数据的存储、存储管理（block大小固定方便计算）、容错等。 读写流程
写入：client端调用filesystem的create方法，后者通过RPC调用NN的create方法，在其namespace中创建新的文件。NN会检查该文件是否存在、client的权限等。成功时返回FSDataOutputStream对象。client对该对象调用write方法，这个对象会选出合适存储数据副本的一组datanode，并以此请求DN分配新的block。这组DN会建立一个管线，例如从client node到最近的DN_1，DN_1传递自己接收的数据包给DN_2。DFSOutputStream自己还有一个确认队列。当所有的DN确认写入完成后，client关闭输出流，然后告诉NN写入完成。
读取：client端通过DistributedFileSystem对象调用open方法，同样通过RPC调用远程的NN方法获取所要查询的文件所涉及的blocks所存储的DN位置，而且这些位置是按照距离排序的。返回的结果是一个FSDataInputStream对象，对输入流对象调用read方法。输入流会从距离最近的DN中读取数据，将数据传递到client，读取结束后关闭流。 这个机制看上去是很笨重的，有了这个分布式文件系统的基础，其他组件就能利用这个系统提供的 API 来对数据的存储进行优化。在介绍下一个组件前，先对主要的主键索引作简单的介绍。 索引 哈希 SSTables/LSM树 BTree/B+Tree 内存：有序集合，例如红黑树、平衡二叉树、跳跃表。 数据结构：平衡多叉树。写入：通过二分查找找到相应的叶子结点进行修改。读取：同上。 MongoDB、Elasticsearch、HBase
主要的主键索引有哈希、LSM、BTree。下面主要涉及到LSM树，所以哈希和BTree这里就不多说了。LSM树有内存和磁盘两个部分....，以跳跃表为例，大致的模型如下图
内存的 MemStore 是一个有序集合，数据写入会先写入这里，当大小达到阈值就会 flush 到磁盘。而后台会有程序按一定策略对这些文件进行合并。合并的原因有：减少小文件，进而减少读取时IO来提升读性能。数据合并，比如图中第二个file有数据a，但现在客户端发送请求要把它删掉或进行修改，如果每次删改都要把数据找到再调整，就会有大量的磁盘IO，所以这些操作一般只做标记，等到后续文件合并时才真正对数据进行修改。还有一个原因是调整排序，因为flush后数据只在file内部有序，合并能够调整整体排序。正因为这种结构，所以LSM的写入是很快的，范围读取也快，因为数据已经有序。而为了保证不读取到旧版本的数据，所以读取需要从最新的开始遍历，这也导致读取旧数据的效率较低。当然，这里面还能优化，但细节就不说了。 简介 HBase 就是基于 HDFS API 构建的一个可以在线低延迟访问大数据的NoSQL数据库。本质上就是给 HDFS 加上一个 LSM Tree 索引，从而提高读写性能。当然，即便优化了，这个高性能也是相对大数据量而言。实际上“HBase并不快，只是当数据量很大的时候它慢的不明显”。由于是 NoSQL 数据库，所以它有文档型数据库的弱项，即基本不支持表关联。 特点 数据量大，单表至少超千万。对稀疏数据尤其适用，因为文档型数据库的 null 就相当于整个字段没有，是不需要占用空间的。 读取近期小范围数据，效率较高，大范围需要计算引擎支持。 场景 时序数据：HBase之上有OpenTSDB模块，可以满足时序类场景的需求 推荐画像：特别是用户的画像，是一个比较大的稀疏矩阵，蚂蚁的风控就是构建在HBase之上 消息/订单等历史数据：在电信领域、银行领域，不少的订单查询底层的存储，另外不少通信、消息同步的应用构建在HBase之上 架构原理
这里大概有10台机器或节点，5个DataNode、两个RegionServer、一个Client、Master、ZooKeeper Client：发送DML、DDL请求，即数据的增删改查和表定义等操作。 实现Master的高可用：当active master宕机，会通过选举机制选取出新master。 管理系统元数据：比如正常工作的RegionServer列表。 辅助RS的宕处理：发现宕机，通知master处理。 分布式锁：方式多个client对同一张表进行表结构修改而产生冲突。 RegionServer 数据的负载均衡、宕机恢复等 RegionServer：处理 client 和 Master 的请求，由 WAL、BlockCache 以及多个 Region 构成。 Store：一个Store存储一个列簇，即一组列。 MemStore和HFile：写缓存，阈值为128M，达到阈值会flush成HFile文件。后台有程序对这些HFile进行合并。 HLog（WAL）：提高数据可靠性。写入数据时先按顺序写入HLog，然后异步刷新落盘。这样即便 MemoStore 的数据丢失，也能通过HLog恢复。而HBase数据的主从复制也是通过HLog回放实现的。 Region：数据表的一个分片，当数据表大小达到一定阈值后会“水平切分”成多个Region，通常同一张表的Region会分不到不同的机器上。 读写过程
client 根据待写入数据的主键（rowkey）寻找合适的 RegionServer 地址，如果没有符合的，就向 zookeeper 查询存储HBase元数据表的 RegionServer 地址。 client 从第一步找到的 RegionServer 查询HBase元数据表，找出合适的写入地址。 将数据写入对应的 RegionServer 的 Region。 简介 Elastic Stack 是以 Elasticsearch 为中心开发的一组组件，其中Kibana、Logstash、Beats使用较多。 Beats 是用 GO 实现的一个开源的用来构建轻量级数据汇集组件，可用于将各种类型的数据发送至 Elasticsearch 与 Logstash。 Logstash：流入、流出 Elasticsearch 的传送带。其他MQ或计算引擎也可以导入ES。 利用 Logstash 同步 Mysql 数据时并非使用 binlog，而且不支持同步删除操作。 Kibana 是 ES 大数据的图形化展示工具。集成了 DSL 命令行查看、数据处理插件、继承了 x-pack（收费）安全管理插件等。 Elasticsearch 搜索引擎，它并不是基于 HDFS 建立的，而是自己实现了分布式存储，并通过各种索引和压缩技术来提高搜索的性能。当然，它作为文档型数据库，其在内存组织数据的方式也是类似LSM树的。 特点 场景 框架原理
Master：主要负责集群中索引的创建、删除以及数据的Rebalance等操作。 Index：一组形成逻辑数据存储的分片的集合，数据库 Shard：Lucene 索引，用于存储和处理 Elasticsearch 索引的一部分。 Segment：Lucene 段，存储了 Lucene 索引的一部分且不可变。结构为倒排索引。 Document：条记录，用以写入 Elasticsearch 索引并从中检索数据。 增删改查原理
Update = Delete + (Index - Ingest Pipeline）
细节补充 倒排索引
一般正向的就是通过文档id找相应的值，而倒排索引则是通过值找文档id。通过倒排这种结构，判断哪些文档包含某个关键词时，就不需要扫描所有文档里面的值，而是从这个关键词列表中去搜索即可。而频率主要是用来计算匹配程度的，默认使用TF-IDF算法。 为什么全文检索中 ES 比 Mysql 快？ Mysql 的辅助索引对于只有一个单词的字段，查询效率就跟 ES 差距不大。 select field1, field2 但如果该字段是有多个单词，那么缺乏分词的 Mysql 就无法建立有效的索引，且查询局限于右模糊，对于“%word%”的搜索效率是极低的。而 ES 通过分词，仍然可以构建出 term dictionary。
然而 Term Dictionary 和 Position 加起来是很大的，难以完全存储在内存。因此，在查找 Term Dictionary 的过程会涉及磁盘IO，效率就会降低。为此，Luence 增加了 term index。这一层通过 Lucene 压缩算法，使得整个 Term Index 存储在内存成为可能。搜索时在内存找到相应的节点，然后再到 Term Dictionary 找即可，省去大量磁盘IO。 内存消耗大 ES 之所以快，很大程度是依赖 Lucene 的缓存以及缓存中的索引结构。而这些缓存只有被预先加载到内存才能做到快速的响应，查询没有被加载的数据通常都是比较慢的，这是 ES 需要大量内存的原因之一。所以有人建议 ES 仅作为内存索引库，即与where、group by、in、sort等过滤、聚合相关的才存储到 ES，而且其他字段并不能帮助查询，只会浪费内存空间。而查询得出的id将返回通过 Mysql 或者 Hbase 进行第二次的查询。由于是主键的搜索，所以不会耗费太多时间。 而 ES 由于给了大部分内存到 Lucene 缓存，那自己聚合计算时用的内存空间就很有限了，这也是 ES 需要大量内存的原因。 目前触漫 ES 情况 刚刚起步，仅仅作为优化部分慢sql查询的解决方案。而 ES 更强大的准实时数据分析、文本搜索功能并没有开发。这其中有涉及到搜索优化（排序规则、分词等）、Kibana可视化、数据冷热分离、各种配置等，所以是需要一定的人力去学习和调试才能发挥它的潜能。 从上面的介绍我们可以知道，ES 是不支持关联的，而且聚合计算的资源很有限。那这时就用到计算引擎了。 计算引擎目前主流的两个开源组件分别是 Spark 和 Flink。从两个引擎的处理模型来看，Spark 的批处理更为高效，Flink 则善于流处理，尽管两者都向着流批一体化的方向发展。当然，只要对弱项做优化还是可以跟另一方未做太多优化的强项比的，只是实现难度大些和效果上限可能略低。比如 Blink，阿里内部的 Flink，其 ML 模块经过优化，在大部分常用模型的计算效率都能高于开源的 Spark 的。如果开源 Spark 也经过阿里那样深度的优化，两者的差距就难说了。 适合：大批量数据的灵活计算，包括关联、机器学习、图计算、实时计算等。 Spark 下面首先介绍 Spark，它是一个用于大规模数据处理的统一分析引擎，其内部主要由 Scala 实现。Spark 当初引起关注主要是它与 Hadoop 的三大件之一的 MapReduce 之间的比较。Hadoop 的三大组件包括 HDFS、Yarn 和 MapReduce。他们三个都是可以拆分开来单独使用的。比如 Yarn 作为资源调度系统，传统 Spark 和 Flink 都会借助它的功能实现任务的调度。而 MapReduce 作为计算引擎，其计算速度当时是弱于 Spark 的，主要是 Spark 减少了不必要的磁盘IO；增加迭代计算功能，从而更好支持机器学习；引入了一些自动优化功能。另外，Spark 广泛的语言支持、API 更强的表达能力等优点都让 Spark 在当时的离线计算领域中超越 MapReduce。 4大场景：Spark 的高层组件包括Spark SQL、Spark Streaming、Spark ML、GraphX。他们都是通过底层组件为 Spark Core 实现具体功能的。但是在使用 Spark 的时候，尽量是不要使用 Spark Core，因为高层组件的产生的 Spark Core一般会更高效，因为Spark做了不少优化，具体后面再说。 多种语言：支持 Java、Python、R 和 Scala 来编写应用代码。 多种部署模式：本地、独立部署、Mesos、Yarn、K8S 多种数据源：HDFS、HBase、Hive、Cassandra、Kafka等 架构原理
Driver 是启动 Spark 作业的JVM进程，它会运行作业（Application）里的main函数，并创建 SparkContext 对象。这个 SparkContext 里面包含这次 Spark 计算的各种配置信息。Spark 通过它实现与 Cluster Manager 通信来申请计算资源。这里的 Cluster Manager，在生产环境一般是 Mesos、Yarn 或者 K8s。这些 Manager 根据其管理的集群情况，给这个 Spark 任务分配相应的容器container，在容器中启动 executor 进程。这些启动后的 executor 会向 Driver 注册，之后 Driver 就可以把它根据用户计算代码生成出的计算任务task发送给这些 executor 执行。计算结束后，结果可能输出到 Driver，也可能输出到当前 executor 的磁盘，或者其他存储。 作业例子 object SparkSQLExample { def main(args: Array[String]): Unit = { // 创建 SparkSession，里面包含 sparkcontext val spark = SparkSession .builder() .appName("Spark SQL basic example") .getOrCreate() import spark.implicits._ // 读取数据 val df1 = spark.read.load("path1...") val df2 = spark.read.load("path2...") // 注册表 df1.createOrReplaceTempView("tb1") df2.createOrReplaceTempView("tb2") // sql val joinedDF = sql( """ |select tb1.id, tb2.field |from tb1 inner join tb2 |on tb1.id = tb2.id """.stripMargin) // driver 终端显示结果 joinedDF.show() // 退出 spark spark.stop() } } SQL会经过一层层的解析然后生成对应的 Java 代码来执行。
与 HBase、 es 和传统数据库查询比较，计算引擎的优势：1）数据量大时速度快，2）计算更加灵活。 文档型数据库：大部分都不支持关联，因为效率低。关联基本都要全文档扫描。因为文档是 schemaless 的，并不确定某个文档是否有关联所需字段。而且个文档的读取都是整个对象的读取，并不会只读某个字段来减少内存开销。另外，这两个组件在内存中本身就有各自的数据结构来服务读写，所以额外的内存用于这类大开销计算也是不现实的。因此，HBase 本身只支持简单的过滤，不支持关联。ES 即便支持过滤、聚合，但依然不支持关联。 传统关系型数据库：可以完成较大数据关联，然而效率低，这主要是受到其大量的磁盘 IO、自身服务（读写、事务等、数据同步）的干扰。在真正大数据情况下，这关联还涉及数据在不同机器的移动，数据库需要维持其数据结构，如 BTree，数据的移动效率较低。 基于内存：计算引擎留有大量内存空间专门用于计算，尽量减少磁盘 IO。 具体而言，Spark 提供了三种 Join 执行策略： BroadcastJoin：当一个大表和一个小表进行Join操作时，为了避免数据的Shuffle，可以将小表的全部数据分发到每个节点上。算法复杂度：O(n). ShuffledHashJoinExec：先对两个表进行hash shuffle，然后把小表变成map完全存储到内存，最后进行join。算法复杂度：O(n)。不适合两个表都很大的情况，因为其中一个表的hash部分要全部放到内存。 SortMergeJoinExec：先hash shuffle将两表数据数据相同key的分到同一个分区，然后sort，最后join。由于排序的特性，每次处理完一条记录后只需要从上一次结束的位置开始继续查找。算法复杂度：O(nlogn)，主要来源于排序。适合大表join大表。之所以适合大表，是因为 join 阶段，可以只读取一部分数据到内存，但其中一块遍历完了，再把下一块加载到内存，这样关联的量就能突破内存限制了。 从上面的例子可以看出计算引擎相比于其他组件在计算方面的优势。 数据流动 下面通过一张图，从另一个角度了解 Spark 的运作。
这是一张简单的数据流程图。描述了一个 WorkCount 的数据流向。其主要代码如下：
val textFile = sc.textFile("hdfs://...") val counts = textFile.map(word => (word, 1)) .reduceByKey(_ + _) counts.saveAsTextFile("hdfs://...") 图中同一阶段有多个数据流体现的是并行。中间的 shuffle 是在聚合、关联、全局排序等操作时会出现的。比如这里的 reduceByKey 就是将相同 key 的数据移动到相同的 partition。这样就能对所有的 a 进行加总，从而得出 a 的总数。 上图的任务是一次性的，或者是周期性的，数据的驱动是拉取型的。如果将数据块换成数据流，map 和 reduce 在启动后就一直存在，并接受数据源不断发送过来的信息，那就变成了流计算。即由周期性变为一直处理，从而变为实时处理，由主动拉取变为被动接收的形式。下面就来介绍 Flink 计算引擎。 Flink Flink 同样是分布式的计算引擎，主要基于Java实现，但它的特色主要体现在流式计算。这个引擎流行的主要推手是阿里。阿里在19年初开源了它修改过的 Flink，收购了 Flink 的母公司，并在各种线下技术论坛上推广 Flink，让 Flink 在 19 年的关注度极速上升。 除了在实时计算领域，Flink 在其他领域或许稍微落后于 Spark，毕竟 Spark 发展比较早，其生态比 Flink 要成熟更多。Flink 目前支持 Scala、Java 和 Python 来写任务代码。功能上同样支持批计算、ML、Graph。部署工具、支持的数据源也 Spark 类似。 实时分析/BI指标：比如某天搞活动或新版本上线，需要尽快根据用户情况来调整策略或发现异常。 实时监控：通过实时统计日志数据来尽快发现线上问题。 架构原理 细节补充 和 Spark 一样，Flink 也会根据 SQL 或者业务代码生成 DAG 图，然后将任务划分并发送给不同的节点执行。最大的不同正如之前所说，数据是实时地、一条条或一小批一小批地不断流进这些节点，然后节点输出响应的结果。而在这种场景下，Flink 在一定程度上解决了实时处理中的不少难点。 保证数据刚好被处理一次，即便在计算过程中出现网络异常或者宕机。 event-time处理，即按照数据中的时间作为计算引擎的时间，这样即便数据上报出现一定的延迟，数据仍然可以被划分到对应的时间窗口。而且还能对一定时间内的数据顺序进行修正。 反压机制，即便数据量极大，Flink 也可以通过自身的机制减缓甚至拒绝接收数据，以免程序被压垮。 小红书实时技术 小红书旧的离线框架和我们现在的大数据体系有点类似，都是把埋点数据上报到日志服务，然后进入离线数仓，只是小红书用 Hive，我们用 DataWorks。然后我们同样也有 T+1 的用户画像、BI报表和推荐的训练数据。
而后续的实时框架是这样的
日志服务的埋点数据先进入 Kafka 这一消息队列里面。不太清楚为什么要加上 Kafka 这一中间件，或许当时并没有开源的 日志服务到Flink 的 connecter 吧。但总之，引入 Flink 之后就可以实时累计埋点中的数据，进而产生实时的画像、BI指标和训练数据了。下面介绍一下这个实时归因
如上图所以，用户app屏幕展示了4个笔记，然后就会有4条曝光埋点，而如果点击笔记、点赞笔记以及从笔记中退出都会有相应的埋点。通过这些埋点就可以得出右面两份简单的训练或分析数据。这些数据跟原来已经积累的笔记/用户画像进行关联就能得出一份维度更多的数据，用于实时的分析或模型预测。实时模型训练这一块至少小红书在19年8月都还没有实现。下图是小红书推荐预测模型的演进
那么如何进行实时训练深度学习模型呢？以下是我的一些想法。借助一个阿里的开源框架flink-ai-extended。
如上图所示，这是 flink 的数据流结构图，左边 source 为数据源，然后进过join、udf等算子进行训练样本数据的生成，然后传递给一个 UDTF/FlatMap 算子，这实际上也是一个 Flink 节点，但它里面包含的是 Tensorflow 的训练 worker，而上下也是 Flink 的节点，都是包含了 Tensorflow 训练所需的一些角色，这样数据源源不断地实时进入 TF 模型来完成实时训练。TF 也可以因此借助 Flink 的分布式框架来完成分布式的学习。多台GPU或者CPU或许应该会比一台GPU的训练效率更高。 这个框架同时适用于模型预测，只要把里面的训练角色换成训练完成的 model，也就可以进行实时的预测，而且这里借助 Flink 内部的通信机制，效率应该会比普通的 http 调用要快不少。 本次分享由于时间有限，讲的都是比较浅层的东西，实际上刚刚所说的每一个组件里面包含的内容都不少，都可以作为一个长远的目标去研究和改造。说回分享的主题之一，使用场景。 首先是存储，上述介绍的 HDFS、HBase、ES（ES虽然是搜索引擎，但它也可以在某些方面替代传统关系型数据的功能） 都是适用于 OLAP 场景，即分析推荐而非事务。从公司目前的情况来看，HDFS 基本可以忽略，因为已经有 DataWork，数据的存储暂时不是问题。更多的问题在于数据使用时的性能。HBase 和 ES 作为文档型数据库，适合一对多的数据模型，比如将帖子和其评论作为一个整体来存储。对于多对一、多对多的模型，文档型数据库实际上并不合适，但可以通过合并宽表、应用层关联等方式在一定程度上进行弥补。而如果多对多关系确实复杂、量大、文档型数据库性能无法满足，比如一些大型社交网络，那么可以考虑图数据库。 当决定尝试文档型数据库时，HBase 的特点在于较为快速地查询小范围的新数据，而且这条数据可以很大。ES 的特点则在于快速的全文检索、准实时的数据分析。当然，分析的复杂度是不能跟计算引擎比的，比如关联、机器学习等。但通过合并宽表、各种where、group by操作，还是能满足不少需求的，尤其是应用的搜索功能，ES 实现起来是比较简单的。目前公司并没有应用它的强项，最好由专人负责它的调试，尤其是搜索排序方面。 然后是计算引擎，目前公司用的 MaxCompute 已经能够满足离线计算的各种需求，或者就欠缺实时计算了。但公司目前实时性需求不多而且也不紧急，所以开发一直都没有启动。目前就看明年推荐是否有这样的需求，而且有相应的prd出来了。而考虑到成本和灵活性，自建或许是更好的选择，比如刚刚提到的 Flink + Tensorflow。 欢迎加入 大数据 |数仓技术交流群 。
扫码加技术群，备注： 数据
识别下方二 维码 ，文末点“在看”解锁后回复“999 ”，即可领取最新大数据资料。感觉干货多，记得设为 星标 哦
欢迎 点赞+收藏+转发朋友圈 三连 看完本文有收获？请转发分享给更多人